package com.wei.web.middleware.retry.compensate;

import com.google.common.base.Joiner;
import com.tuya.basic.mq.IMqKafkaBizConsumer;
import com.tuya.basic.mq.domain.KafkaMqData;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.toolkit.trace.Trace;

import javax.annotation.Resource;
import java.util.Arrays;

/**
 * @author shihuimiao
 * create time: 上午9:52
 */
@Slf4j
public abstract class AbstractMqKafkaBizConsumer<T> implements IMqKafkaBizConsumer<T>, Processor<T> {

    @Resource
    CompensateService compensateService;

    /**
     * kafka文档中语句：消费逻辑类抛出异常，默认消息会丢回topic再次消费，返回true表示消费成功，返回false表示虽然消费失败但是可以接受不需要额外做处理。
     *
     * @param topic
     * @param data
     * @param partition
     * @param offset
     * @param key
     * @return
     * @throws Exception
     */
    @Trace
    @Override
    public final boolean consumber(String topic, KafkaMqData<T> data, int partition, long offset, String key) throws Exception {
        log.debug("kafka consumer ; topic={};data={};partition={};offset={},key={}", topic, data, partition, offset, key);
        Boolean isSuccess = null;

        String businessKey = getUniqueKey(topic, data, partition, offset, key);

        try {
            isSuccess = execute(data.getData());
            log.debug("isSuccess : {}", isSuccess);
            return isSuccess;
        } catch (Exception e) {
            log.error("consumber unexpected error !", e);
            if (data.getCount() >= retryTime()) {
                failMqSave(businessKey, data.getData());
            }
            throw e;
        } finally {
            //业务允许放弃；自己补偿
            if (isSuccess != null && !isSuccess) {
                failMqSave(businessKey, data.getData());
            }
        }
    }

    /**
     * 业务方需要自己生成唯一key 复写此方法
     *
     * @param data
     * @return
     */
    public String getUniqueKey(String topic, KafkaMqData<T> data, int partition, long offset, String key) {
        return Joiner.on(":").join(Arrays.asList(topic, partition, offset));
    }

    /**
     * kafka那配置的重试次数 测试得到count值=2就进入死信队列 如果改了配置 需要自己复写这个方法
     *
     * @return
     */
    protected int retryTime() {
        return 2;
    }

    private void failMqSave(String key, T data) {
        try {
            compensateService.saveTask(key, data, this.getClass().getCanonicalName(), timesRequired());
        } catch (Exception e) {
            log.error("structureDO error;key={},data={},class={}", key, data, this.getClass(), e);
        }
    }

    /**
     * 全局补偿次数定义在properties中 也支持自定义补偿次数 如果需要可以不复写这个方法
     *
     * @return
     */
    protected Integer timesRequired() {
        return null;
    }


}
